package SparkStreaming

import org.apache.spark.streaming.dstream.{DStream, ReceiverInputDStream}
import org.apache.spark.streaming.{Milliseconds, StreamingContext}
import org.apache.spark.{SparkConf, SparkContext}

/**
  * Created by Administrator on 2018/5/29.
  */
object StreamingWordCount {
  def main(args: Array[String]): Unit = {
    //
    val conf: SparkConf = new SparkConf().setAppName("StreamingWordCount").setMaster("local[2]")
    //
    val sc: SparkContext = new SparkContext(conf)
    //StreamingContext是对SparkContext的包装，增加了实时的功能
    //第二个参数是小批次产生的时间间隔
    val ssc: StreamingContext = new StreamingContext(sc, Milliseconds(5000))

    //通过调用API创建SparkStreaming的抽象DStream
    //从一个Socket端口中读取数据
    //这一步要先确保在linux主机中用yum安装nc
    val lines: ReceiverInputDStream[String] = ssc.socketTextStream("192.168.80.92", 8888)

    //接下来跟操作RDD一样
    //切分压平
    val words: DStream[String] = lines.flatMap(_.split(" "))
    //单词和一组合在一起
    val wordAndOne: DStream[(String, Int)] = words.map((_, 1))
    //聚合
    val reduced: DStream[(String, Int)] = wordAndOne.reduceByKey(_+_)
    //打印结果（Action）
    reduced.print()

    //启动SparkStreaming程序
    ssc.start()
    //优雅的等待退出
    ssc.awaitTermination()

  }
}
